#include "messaging/kafka_producer.h"
#include "common/logger/logger.h"
#include <librdkafka/rdkafkacpp.h>
#include <map>

using namespace common::logger;

namespace messaging {

// 实现delivery report回调
class DeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    explicit DeliveryReportCb(DeliveryCallback callback) : callback_(callback) {}

    void dr_cb(RdKafka::Message& message) override {
        if (callback_) {
            callback_(message.err() == RdKafka::ERR_NO_ERROR, 
                     message.errstr());
        }
    }

private:
    DeliveryCallback callback_;
};

// 实现错误回调
class ErrorCb : public RdKafka::EventCb {
public:
    void event_cb(RdKafka::Event& event) override {
        switch (event.type()) {
            case RdKafka::Event::EVENT_ERROR:
                LOG_ERROR("Kafka错误: " + event.str());
                break;
            case RdKafka::Event::EVENT_STATS:
                LOG_INFO("Kafka统计: " + event.str());
                break;
            case RdKafka::Event::EVENT_LOG:
                LOG_INFO("Kafka日志: " + event.str());
                break;
            default:
                LOG_INFO("Kafka事件: " + event.str());
                break;
        }
    }
};

// 创建Kafka生产者
std::shared_ptr<KafkaProducer> KafkaProducer::create(const KafkaProducerConfig& config) {
    auto producer = std::shared_ptr<KafkaProducer>(new KafkaProducer(config));
    if (!producer->init(config)) {
        return nullptr;
    }
    return producer;
}

KafkaProducer::KafkaProducer(const KafkaProducerConfig& /* config */) 
    : isRunning_(false) {
}

KafkaProducer::~KafkaProducer() {
    if (producer_) {
        flush();
        topics_.clear();
        producer_.reset();
    }
}

bool KafkaProducer::init(const KafkaProducerConfig& config) {
    try {
        // 创建配置对象
        std::string errstr;
        RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
        
        // 设置基本配置
        conf->set("bootstrap.servers", config.brokers, errstr);
        conf->set("client.id", config.clientId, errstr);
        conf->set("message.timeout.ms", std::to_string(config.timeoutMs), errstr);
        conf->set("queue.buffering.max.ms", std::to_string(config.queueBufferingMaxMs), errstr);
        conf->set("queue.buffering.max.messages", std::to_string(config.queueBufferingMaxMessages), errstr);
        conf->set("queue.buffering.max.kbytes", std::to_string(config.queueBufferingMaxKbytes), errstr);
        conf->set("request.required.acks", std::to_string(config.requestRequiredAcks), errstr);
        
        // 设置错误回调
        auto errorCb = new ErrorCb();
        conf->set("event_cb", errorCb, errstr);
        
        // 创建生产者
        RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);
        if (!producer) {
            LOG_ERROR("创建Kafka生产者失败: " + errstr);
            delete conf; // 如果创建失败，需要手动删除配置对象
            return false;
        }
        
        // 转移所有权
        conf_.reset(conf);
        producer_.reset(producer);
        
        LOG_INFO("Kafka生产者已创建，brokers: " + config.brokers);
        isRunning_ = true;
        return true;
    } catch (const std::exception& e) {
        LOG_ERROR("初始化Kafka生产者异常: " + std::string(e.what()));
        return false;
    }
}

bool KafkaProducer::send(const std::string& topic, const std::string& key, 
                       const std::string& value, DeliveryCallback callback) {
    if (!isRunning_ || !producer_) {
        LOG_ERROR("Kafka生产者未初始化或已停止");
        return false;
    }
    
    try {
        // 获取或创建主题
        auto topicObj = getTopic(topic);
        if (!topicObj) {
            LOG_ERROR("创建Kafka主题失败: " + topic);
            return false;
        }
        
        // 设置delivery report回调
        if (callback) {
            std::string errstr;
            auto deliveryCb = new DeliveryReportCb(callback);
            conf_->set("dr_cb", deliveryCb, errstr);
        }
        
        // 发送消息
        void* valueData = const_cast<char*>(value.c_str());
        size_t valueLen = value.size();
        void* keyData = key.empty() ? nullptr : const_cast<char*>(key.c_str());
        size_t keyLen = key.empty() ? 0 : key.size();
        
        RdKafka::ErrorCode err = producer_->produce(
            topicObj,                          // 主题
            RdKafka::Topic::PARTITION_UA,      // 分区（自动分配）
            RdKafka::Producer::RK_MSG_COPY,    // 消息标志（复制）
            valueData, valueLen,               // 消息值和长度
            keyData, keyLen,                   // 消息键和长度
            nullptr);                          // 消息opaque
        
        if (err != RdKafka::ERR_NO_ERROR) {
            LOG_ERROR("发送Kafka消息失败: " + RdKafka::err2str(err));
            return false;
        }
        
        // 轮询以处理回调
        producer_->poll(0);
        return true;
    } catch (const std::exception& e) {
        LOG_ERROR("发送Kafka消息异常: " + std::string(e.what()));
        return false;
    }
}

std::future<bool> KafkaProducer::sendAsync(const std::string& topic, const std::string& key, 
                                         const std::string& value) {
    auto promise = std::make_shared<std::promise<bool>>();
    auto future = promise->get_future();
    
    auto callback = [promise](bool success, const std::string& /* error */) {
        promise->set_value(success);
    };
    
    if (!send(topic, key, value, callback)) {
        promise->set_value(false);
    }
    
    return future;
}

void KafkaProducer::flush(int timeoutMs) {
    if (producer_) {
        producer_->flush(timeoutMs);
    }
}

void KafkaProducer::poll(int timeoutMs) {
    if (producer_) {
        producer_->poll(timeoutMs);
    }
}

RdKafka::Topic* KafkaProducer::getTopic(const std::string& topic) {
    // 检查主题是否已存在
    auto it = topics_.find(topic);
    if (it != topics_.end()) {
        return it->second.get();
    }
    
    // 创建主题配置
    std::string errstr;
    RdKafka::Conf* topicConf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
    
    // 创建主题
    RdKafka::Topic* topicObj = RdKafka::Topic::create(producer_.get(), topic, topicConf, errstr);
    
    // 手动删除配置对象，它已被Topic::create复制/使用
    delete topicConf;
    
    if (!topicObj) {
        LOG_ERROR("创建Kafka主题失败: " + errstr);
        return nullptr;
    }
    
    auto result = topicObj;
    topics_[topic].reset(topicObj);
    return result;
}

} // namespace messaging 